package com.nageoffer.shortlink.project.mq.consumer;

import cn.hutool.core.date.DateUtil;
import cn.hutool.core.date.Week;
import cn.hutool.core.util.StrUtil;
import cn.hutool.http.HttpUtil;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.nageoffer.shortlink.project.dao.entity.*;
import com.nageoffer.shortlink.project.dao.mapper.*;
import com.nageoffer.shortlink.project.dto.biz.ShortLinkStatsRecordDTO;
import com.nageoffer.shortlink.project.mq.idempotent.MessageQueueIdempotentHandler;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.redisson.api.RLock;
import org.redisson.api.RReadWriteLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Component;

import java.util.Date;
import java.util.HashMap;
import java.util.Map;

import static com.nageoffer.shortlink.project.common.constant.RedisKeyConstant.LOCK_GID_UPDATE_KEY;
import static com.nageoffer.shortlink.project.common.constant.ShortLinkConstant.AMAP_REMOTE_URL;

@Slf4j
@Component
@RequiredArgsConstructor
@RocketMQMessageListener(topic = "short-link-stats-save-topic", consumerGroup = "my-consumer-group")
public class RMQShortLinkStatsSaveConsumer implements  RocketMQListener<Message<Map<String, String>>>{


        private final ShortLinkMapper shortLinkMapper;
        private final ShortLinkGotoMapper shortLinkGotoMapper;
        private final RedissonClient redissonClient;
        private final LinkAccessStatsMapper linkAccessStatsMapper;
        private final LinkLocaleStatsMapper linkLocaleStatsMapper;
        private final LinkOsStatsMapper linkOsStatsMapper;
        private final LinkBrowserStatsMapper linkBrowserStatsMapper;
        private final LinkAccessLogsMapper linkAccessLogsMapper;
        private final LinkDeviceStatsMapper linkDeviceStatsMapper;
        private final LinkNetworkStatsMapper linkNetworkStatsMapper;
        private final LinkStatsTodayMapper linkStatsTodayMapper;
        private final StringRedisTemplate stringRedisTemplate;
        private final MessageQueueIdempotentHandler messageQueueIdempotentHandler;


        @Value("${short-link.stats.locale.amap-key}")
        private String statsLocaleAmapKey;


        @Override
        public void onMessage(Message<Map<String, String>> message) {
            //获取消息的key
            String keys = (String) message.getHeaders().get("KEYS");
            //todo 缓存中，判断消息的状态
            System.out.println("消息：" + keys + "是第一次消费");
            //获取消息体
            Map<String, String> payload = message.getPayload();
            String statsRecord = payload.get("statsRecord");
            ShortLinkStatsRecordDTO recordDTO = JSONObject.parseObject(statsRecord, ShortLinkStatsRecordDTO.class);
            actualSaveShortLinkStats(recordDTO);
            //todo 从缓存中删除消息or设置消息状态
            System.out.println("消息：" + keys + "已经消费完成");
        }

        /**
         * 实际保存短链接监控状态
         * 这里使用读写锁，因为更新短链接访问量的操作需要传入gid，当gid修改的时候，需要加锁，避免并发修改gid导致数据不一致。
         * 而当gid不修改的时候，使用读锁可以有效提高并发量。
         *
         * 为什么不能直接使用普通的分布式锁？
         * 因为短链接的gid是可修改的，使用普通的分布式锁，可能会导致统计的数据错误。
         *
         */
        public void actualSaveShortLinkStats(ShortLinkStatsRecordDTO statsRecord) {
            //获取完整短链接
            String fullShortUrl = statsRecord.getFullShortUrl();
            //使用读写锁
            RReadWriteLock readWriteLock = redissonClient.getReadWriteLock(String.format(LOCK_GID_UPDATE_KEY, fullShortUrl));
            //获取读锁
            RLock rLock = readWriteLock.readLock();
            rLock.lock();
            try {
                //通过短链接查gid
                LambdaQueryWrapper<ShortLinkGotoDO> queryWrapper = Wrappers.lambdaQuery(ShortLinkGotoDO.class)
                        .eq(ShortLinkGotoDO::getFullShortUrl, fullShortUrl);
                ShortLinkGotoDO shortLinkGotoDO = shortLinkGotoMapper.selectOne(queryWrapper);
                String gid = shortLinkGotoDO.getGid();
                //获取短链接的访问时间，时、星期
                Date currentDate = statsRecord.getCurrentDate();
                int hour = DateUtil.hour(currentDate, true);
                Week week = DateUtil.dayOfWeekEnum(currentDate);
                int weekValue = week.getIso8601Value();
                //pv：page view访问量；uv：unique visitor独立访客；uip：unique ip独立ip
                LinkAccessStatsDO linkAccessStatsDO = LinkAccessStatsDO.builder()
                        .pv(1)
                        .uv(statsRecord.getUvFirstFlag() ? 1 : 0)
                        .uip(statsRecord.getUipFirstFlag() ? 1 : 0)
                        .hour(hour)
                        .weekday(weekValue)
                        .fullShortUrl(fullShortUrl)
                        .date(currentDate)
                        .build();
                linkAccessStatsMapper.shortLinkStats(linkAccessStatsDO);
                //统计地域
                //利用工具类，通过ip获取访问地址。城市、省份。
                Map<String, Object> localeParamMap = new HashMap<>();
                localeParamMap.put("key", statsLocaleAmapKey);
                localeParamMap.put("ip", statsRecord.getRemoteAddr());
                String localeResultStr = HttpUtil.get(AMAP_REMOTE_URL, localeParamMap);
                JSONObject localeResultObj = JSON.parseObject(localeResultStr);
                String infoCode = localeResultObj.getString("infocode");
                String actualProvince = "未知";
                String actualCity = "未知";
                if (StrUtil.isNotBlank(infoCode) && StrUtil.equals(infoCode, "10000")) {
                    String province = localeResultObj.getString("province");
                    boolean unknownFlag = StrUtil.equals(province, "[]");
                    LinkLocaleStatsDO linkLocaleStatsDO = LinkLocaleStatsDO.builder()
                            .province(actualProvince = unknownFlag ? actualProvince : province)
                            .city(actualCity = unknownFlag ? actualCity : localeResultObj.getString("city"))
                            .adcode(unknownFlag ? "未知" : localeResultObj.getString("adcode"))
                            .cnt(1)
                            .fullShortUrl(fullShortUrl)
                            .country("中国")
                            .date(currentDate)
                            .build();
                    linkLocaleStatsMapper.shortLinkLocaleState(linkLocaleStatsDO);
                }
                //统计操作系统
                LinkOsStatsDO linkOsStatsDO = LinkOsStatsDO.builder()
                        .os(statsRecord.getOs())
                        .cnt(1)
                        .fullShortUrl(fullShortUrl)
                        .date(currentDate)
                        .build();
                linkOsStatsMapper.shortLinkOsState(linkOsStatsDO);
                //统计浏览器
                LinkBrowserStatsDO linkBrowserStatsDO = LinkBrowserStatsDO.builder()
                        .browser(statsRecord.getBrowser())
                        .cnt(1)
                        .fullShortUrl(fullShortUrl)
                        .date(currentDate)
                        .build();
                linkBrowserStatsMapper.shortLinkBrowserState(linkBrowserStatsDO);
                //统计设备
                LinkDeviceStatsDO linkDeviceStatsDO = LinkDeviceStatsDO.builder()
                        .device(statsRecord.getDevice())
                        .cnt(1)
                        .fullShortUrl(fullShortUrl)
                        .date(currentDate)
                        .build();
                linkDeviceStatsMapper.shortLinkDeviceState(linkDeviceStatsDO);
                //统计网络
                LinkNetworkStatsDO linkNetworkStatsDO = LinkNetworkStatsDO.builder()
                        .network(statsRecord.getNetwork())
                        .cnt(1)
                        .fullShortUrl(fullShortUrl)
                        .date(currentDate)
                        .build();
                linkNetworkStatsMapper.shortLinkNetworkState(linkNetworkStatsDO);
                //统计访问日志
                LinkAccessLogsDO linkAccessLogsDO = LinkAccessLogsDO.builder()
                        .user(statsRecord.getUv())
                        .ip(statsRecord.getRemoteAddr())
                        .browser(statsRecord.getBrowser())
                        .os(statsRecord.getOs())
                        .network(statsRecord.getNetwork())
                        .device(statsRecord.getDevice())
                        .locale(StrUtil.join("-", "中国", actualProvince, actualCity))
                        .fullShortUrl(fullShortUrl)
                        .build();
                linkAccessLogsMapper.insert(linkAccessLogsDO);
                //更新短链接的访问量
                shortLinkMapper.incrementStats(gid, fullShortUrl, 1, statsRecord.getUvFirstFlag() ? 1 : 0, statsRecord.getUipFirstFlag() ? 1 : 0);
                LinkStatsTodayDO linkStatsTodayDO = LinkStatsTodayDO.builder()
                        .todayPv(1)
                        .todayUv(statsRecord.getUvFirstFlag() ? 1 : 0)
                        .todayUip(statsRecord.getUipFirstFlag() ? 1 : 0)
                        .fullShortUrl(fullShortUrl)
                        .date(currentDate)
                        .build();
                //更新今日访问量
                linkStatsTodayMapper.shortLinkTodayState(linkStatsTodayDO);
            } catch (Throwable ex) {
                log.error("短链接访问量统计异常", ex);
            } finally {
                rLock.unlock();
            }
        }


}

